package com.ls.fw.data.search.impl.dao.base;

import java.util.Arrays;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthStatus;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.flush.FlushResponse;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollResponse;
import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ls.fw.data.exception.DataException;

public abstract class ESBaseDaoAdapter implements ESBaseDao{

	public Logger logger = LoggerFactory.getLogger(this.getClass());
	
	public TransportClient server = null;

    public ESBaseDaoAdapter() {
		super();
	}

	public ESBaseDaoAdapter(TransportClient client) {
		super();
		this.server = client;
	}

	/**
     * Ensures the cluster has a green state via the cluster health API. This method will also wait for relocations.
     * It is useful to ensure that all action on the cluster have finished and all shards that were currently relocating
     * are now allocated and started.
     */
	public ClusterHealthStatus  ensureGreen(String... indices) {
        return ensureGreen(TimeValue.timeValueSeconds(30), indices);
    }

    /**
     * Ensures the cluster has a green state via the cluster health API. This method will also wait for relocations.
     * It is useful to ensure that all action on the cluster have finished and all shards that were currently relocating
     * are now allocated and started.
     *
     * @param timeout time out value to set on {@link org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest}
     */
    public ClusterHealthStatus ensureGreen(TimeValue timeout, String... indices) {
        ClusterHealthResponse actionGet = server.admin().cluster()
                .health(Requests.clusterHealthRequest(indices).timeout(timeout).waitForGreenStatus().waitForEvents(Priority.LANGUID).waitForRelocatingShards(0)).actionGet();
        if (actionGet.isTimedOut()) {
            logger.info("ensureGreen timed out, cluster state:\n{}\n{}", server.admin().cluster().prepareState().get().getState().prettyPrint()
            		, server.admin().cluster().preparePendingClusterTasks().get().prettyPrint());
        }
        logger.debug("indices {} are green", indices.length == 0 ? "[_all]" : indices);
        return actionGet.getStatus();
    }

    /**
     * Waits for all relocating shards to become active using the cluster health API.
     */
    public ClusterHealthStatus waitForRelocation() {
        return waitForRelocation(null);
    }

    /**
     * Waits for all relocating shards to become active and the cluster has reached the given health status
     * using the cluster health API.
     */
    public ClusterHealthStatus waitForRelocation(ClusterHealthStatus status) {
        ClusterHealthRequest request = Requests.clusterHealthRequest().waitForRelocatingShards(0);
        if (status != null) {
            request.waitForStatus(status);
        }
        ClusterHealthResponse actionGet = server.admin().cluster()
                .health(request).actionGet();
        if (actionGet.isTimedOut()) {
            logger.info("waitForRelocation timed out (status={}), cluster state:\n{}\n{}"
            		, status, server.admin().cluster().prepareState().get().getState().prettyPrint()
            		, server.admin().cluster().preparePendingClusterTasks().get().prettyPrint());
        }
        return actionGet.getStatus();
    }

    
    /**
     * Ensures the cluster is in a searchable state for the given indices. This means a searchable copy of each
     * shard is available on the cluster.
     */
    public ClusterHealthStatus ensureSearchable(String... indices) {
        // this is just a temporary thing but it's easier to change if it is encapsulated.
        return ensureGreen(indices);
    }

    /**
     * Syntactic sugar for:
     * <pre>
     *   client.prepareIndex(index, type).setSource(source).execute().actionGet();
     * </pre>
     */
    public final IndexResponse index(String index, String type, XContentBuilder source) {
        return server.prepareIndex(index, type).setSource(source).execute().actionGet();
    }

    /**
     * Syntactic sugar for:
     * <pre>
     *   client.prepareIndex(index, type).setSource(source).execute().actionGet();
     * </pre>
     */
    public final IndexResponse index(String index, String type, String id, Map<String, Object> source) {
        return server.prepareIndex(index, type, id).setSource(source).execute().actionGet();
    }

    /**
     * Syntactic sugar for:
     * <pre>
     *   client.prepareGet(index, type, id).execute().actionGet();
     * </pre>
     */
    public final GetResponse get(String index, String type, String id) {
        return server.prepareGet(index, type, id).execute().actionGet();
    }

    /**
     * Syntactic sugar for:
     * <pre>
     *   return client.prepareIndex(index, type, id).setSource(source).execute().actionGet();
     * </pre>
     */
    public final IndexResponse index(String index, String type, String id, XContentBuilder source) {
        return server.prepareIndex(index, type, id).setSource(source).execute().actionGet();
    }

    /**
     * Syntactic sugar for:
     * <pre>
     *   return client.prepareIndex(index, type, id).setSource(source).execute().actionGet();
     * </pre>
     */
    public final IndexResponse index(String index, String type, String id, Object... source) {
        return server.prepareIndex(index, type, id).setSource(source).execute().actionGet();
    }

    /**
     * Syntactic sugar for:
     *
     * <pre>
     *   return client.prepareIndex(index, type, id).setSource(source).execute().actionGet();
     * </pre>
     *
     * where source is a String.
     */
    public final IndexResponse index(String index, String type, String id, String source) {
        return server.prepareIndex(index, type, id).setSource(source).execute().actionGet();
    }

    /**
     * Waits for relocations and refreshes all indices in the cluster.
     *
     * @see #waitForRelocation()
     */
    public final RefreshResponse refresh() {
        waitForRelocation();
        // TODO RANDOMIZE with flush?
        RefreshResponse actionGet = server.admin().indices().prepareRefresh().execute().actionGet();
        return actionGet;
    }

    /**
     * Flushes and refreshes all indices in the cluster
     */
    public final void flushAndRefresh() {
        flush(true);
        refresh();
    }

    /**
     * Flushes all indices in the cluster
     */
    public final FlushResponse flush() {
        return flush(true);
    }

    public FlushResponse flush(boolean ignoreNotAllowed) {
        waitForRelocation();
        FlushResponse actionGet = server.admin().indices().prepareFlush().setWaitIfOngoing(true).execute().actionGet();
        if (ignoreNotAllowed) {
           this.showErrorInfo(actionGet);
        } else {
        	if(actionGet.getShardFailures() != null && actionGet.getShardFailures().length>0){
        		throw new DataException(Arrays.toString(actionGet.getShardFailures()));
        	}
        }
        return actionGet;
    }
    
    public void showErrorInfo(BroadcastOperationResponse response){
    	for (ShardOperationFailedException failure : response.getShardFailures()) {
            this.logger.info("unexpected flush failure " + failure.reason());
        }
    }

    /**
     * Returns <code>true</code> iff the given index exists otherwise <code>false</code>
     */
    public boolean indexExists(String index) {
        IndicesExistsResponse actionGet = server.admin().indices().prepareExists(index).execute().actionGet();
        return actionGet.isExists();
    }

    /**
     * Returns a random admin client. This client can either be a node or a transport client pointing to any of
     * the nodes in the cluster.
     */
    public AdminClient admin() {
        return server.admin();
    }
    

    /**
     * Clears the given scroll Ids
     */
    public boolean clearScroll(String... scrollIds) {
        ClearScrollResponse clearResponse = server.prepareClearScroll()
                .setScrollIds(Arrays.asList(scrollIds)).get();
        return clearResponse.isSucceeded();
    }

	/**
	 * @param server the server to set
	 */
	public void setServer(TransportClient server) {
		this.server = server;
	}

	@SuppressWarnings("unchecked")
	@Override
	public TransportClient getServer() {
		return server;
	}
	
	
	
	/**
	 * 创建索引库
	 * @author ls
	 * 2014年12月4日下午3:47:19 
	 * @param index
	 * @param aliase
	 * @return
	 */
	public boolean prepareCreate(String index,String aliase){
		if(StringUtils.isBlank(aliase)){
			 throw new DataException("别名不能为空！");
		}
		boolean exist = server.admin().indices().prepareExists(index)
				.execute().actionGet().isExists();
		if(!exist){
			 CreateIndexRequestBuilder builder = server.admin().indices().prepareCreate(index);
			 builder.addAlias(new Alias(aliase));
			 exist = builder.get().isAcknowledged();
		}else{
			exist = this.addAliase(index, aliase);
		}
		return exist;
	}
	
	
	
	
	public boolean addAliase(String index, String aliase){
		if(StringUtils.isBlank(aliase)){
			 throw new DataException("别名不能为空！");
		}
		this.logger.info("addAliase:"+index+"："+aliase);
		if(!this.exist(index, aliase)){
			return admin().indices().prepareAliases().addAlias(index, aliase).get().isAcknowledged();
		}
		return false;
	}
	
	
	public boolean exist(String index, String aliase){
		if(StringUtils.isBlank(aliase)){
			 throw new DataException("别名不能为空！");
		}
		this.logger.info("existAliase:"+index+"："+aliase);
		GetAliasesRequest request = new GetAliasesRequest(aliase);
		request.indices(index);
		return admin().indices().aliasesExist(request).actionGet().exists();
	}
	
	public boolean removeAliase(String index, String aliase){
		if(StringUtils.isBlank(aliase)){
			 throw new DataException("别名不能为空！");
		}
		this.logger.info("removeAliase:"+index+"："+aliase);
		if(this.exist(index, aliase)){
			return admin().indices().prepareAliases().removeAlias(index, aliase).get().isAcknowledged();
		}
		return false;
	}
	
	public boolean removeAliase(String removeindex, String newindex, String aliase){
		if(StringUtils.isBlank(aliase)){
			 throw new DataException("别名不能为空！");
		}
		this.logger.info("removeAliase:"+removeindex+"-->"+newindex+"："+aliase);
		if(this.exist(removeindex, aliase)){
			return admin().indices().prepareAliases().removeAlias(removeindex, aliase)
					.addAlias(newindex, aliase).get().isAcknowledged();
		}else{
			return this.addAliase(newindex, aliase);
		}
	}
}
